/**
 * Created by IntelliJ IDEA.
 * User: jiyufei
 * Date: 2018/10/17
 * Time: 14:20
 **/
package com.jiyufei.linux.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;
import com.jiyufei.linux.bean.OtherLog;
import com.jiyufei.linux.log.EsLogCenter;
import com.jiyufei.linux.util.ElasticSearchInfo;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.config.HttpClientConfig;
import io.searchbox.core.Index;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

public class OtherBolt extends BaseRichBolt {
    
    private OutputCollector outputCollector;
    private TopologyContext topologyContext;
    private JestClient jestClient;
    private Index index;
    private EsLogCenter esLogCenter;
    private SimpleDateFormat sdf;
    
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.topologyContext = topologyContext;
        this.outputCollector = outputCollector;
        JestClientFactory jestClientFactory = new JestClientFactory();
        jestClientFactory.setHttpClientConfig(new HttpClientConfig.Builder(ElasticSearchInfo.ES_IP).multiThreaded(true).build());
        jestClient = jestClientFactory.getObject();
        esLogCenter = new EsLogCenter();
        sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZZ");
    }
    
    @Override
    public void execute(Tuple tuple) {
        String log = tuple.getStringByField("log");
        OtherLog otherLog = new OtherLog();
        otherLog.setTime(sdf.format(new Date()));
        otherLog.setType("other");
        otherLog.setLog(log);
        try {
            index = new Index.Builder(otherLog).index("yunying-other").type(otherLog.getType()).build();
            jestClient.execute(index);
            outputCollector.ack(tuple);
        }catch (Exception e){
            esLogCenter.addLog("other",topologyContext.getTopologyId() + ":" + topologyContext.getThisComponentId(),"yunying",
                    "[HlasLinuxLog.OtherBolt]","High","类型匹配失败。log:" + log);
            outputCollector.fail(tuple);
        }
    }
    
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    
    }
}
